#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <ctype.h>
#include <sys/stat.h>
#include <dirent.h>

#include <iostream>
#include <string.h>
#include <assert.h>
#include <map>
#include <fstream>
#include <set>
#include <vector>

#include "mpi.h"

using namespace std;

extern char *optarg;
extern int optind;
extern int optopt;
extern int opterr;

//we assign a default value for N
int N = 3;

//the default directory is a empty string.
char dir[1024] = "";


int rank; //the rank of me(i.e., this process)
int size; //the number of processes in all


int to = 0;   //the destination rank
int from = 0; //the source rank
int count = 0;//the count of elements to be sent/received
int tag = 0;  //the tag of the message

int st_count, st_source, st_tag; //the information of the message that actually recieved.
MPI::Status status; //the status of the received message

char buffer[1<<14]; //the buffer used to store the file names send from rank 0 to all the ranks

//the vector containing filenames
vector<string> filepaths;

//
class Word
{
public:
	string word;
	int count;
public:
	Word(void) : count(1) {}

	bool operator < (const Word & word) const
	{
		if(this->count > word.count)
			return true;
		else if (this->count < word.count)
			return false;
		else	
			return (this->word).compare(word.word) <= 0 ? true : false;
	}
};

//This is the dictionary we maintain in order to store the map between each distinct word and its number of occurance in a file.
class Dict
{
public:
	map<string, Word> dict;
	string filename;
public:
	Dict(string fn) : filename(fn) {}
	void Add(char * word);
	void MostCommonWords(int n = N);
	~Dict();
};

//store the words that occur most frequently in a set and print them
void Dict::MostCommonWords(int n)
{
	set <Word> words;
	//generate a buffere to store most frequent words (the parameter of -n) in it
	strcpy(buffer, "");

	for(map<string, Word>::iterator i = dict.begin(); i != dict.end(); i++)
		words.insert(i->second);//insert it in the set

	//Each entry of the message contains three parts, the belonged filename, the word, and
	////its count
	////construct the message to be sent
	for(set <Word>::iterator i = words.begin(); i != words.end(); ++i)
	{
		if(--n >= 0)
		{
			char temp[1024];
			//construct the message to be sent

			sprintf(temp, "%s,%s,%d\n", filename.c_str(), i->word.c_str(), i->count);
			strcat(buffer, temp);
		}
		else
			break;
	}

}

//This funciton add a word into a dictionary
void Dict::Add(char * word)
{
	//if this word is empty, do nothing
	if(strcmp(word, "") == 0)
		return;
	
	//change the word into lowercase word
  	for (int i = 0; word[i]; i++)
	    word[i] = tolower(tolower(word[i]));

	//find the place where the word is located in the dictionary.
	string word_str(word);
	map<string, Word>::iterator map_ptr;
	map_ptr = dict.find(word_str);
        //if this word is not found, insert the new word, esle we increment its frequency	
	if(map_ptr != dict.end())
	{
		Word & word = dict[word_str];
		word.count++;
	}
	else
	{
		Word * word = new Word();
		word->word = word_str;
		dict[word_str] = *word;
	}
}

//deconstructing function
Dict::~Dict()
{
	for(map<string, Word>::iterator wi = dict.begin(); wi != dict.end(); wi++)
		delete &(wi->second);
}

//for each file, we process the file
void process_this_file(string filename)
{

	fstream file(filename.c_str(), ios::in);
	if(!file.is_open())
	{

		cout<<"No File"<<filename;
		cerr<<"file "<<filename<<" cannot be open!";
		return;
	}

	//create a new dictionary
	Dict * pdict = new Dict(filename);

	string buffer;
        //process each line, from which we split words according to the punctuaions.	
	while(getline(file, buffer))
	{
		char * c_buffer = new char [buffer.size() + 1];
		strcpy(c_buffer, buffer.c_str());

		char * word = strtok(c_buffer, " \t\n\v\f\r.?!:;-()[]'\"/,");

		while(word)
		{
			pdict->Add(word);
			word = strtok(NULL, " \t\n\v\f\r.?!:;-()[]'\"/,");
		}

		delete [] c_buffer;
	}

        //calcuate the most N frequent occuring words in this dictionary
	pdict->MostCommonWords(N);

	delete pdict;
}

//Send the reslt to the rank 0
void SendResult(char * buffer)
{
	assert(buffer);

	to = 0;
	tag = st_tag;
	
	count = strlen(buffer) + 1;
	MPI::COMM_WORLD.Send(buffer, count, MPI::CHAR, to, tag);
}

//tell other processors to die
void TellOthersToDie(void)
{

	for(int i = 1; i < size; i ++)
		MPI::COMM_WORLD.Send("YOU_CAN_DIE_NOW", 16, MPI::CHAR, i, 0);
}

//the main program
int main(int argc, char ** argv)
{

	MPI::Init(argc, argv);

        //get the rank of this processor, as well as the number of processors in this communcation world.	
	rank = MPI::COMM_WORLD.Get_rank();
	size = MPI::COMM_WORLD.Get_size();

	optind = 1;
	opterr = 0;
	int result;

        //using the getopt function to handle options	
	while((result = getopt(argc, argv, "d:n:")) != -1)
	{
		switch((char)result)
		{
			case 'd':
				strcpy(dir, optarg);
				break;
			case 'n':
				N = atoi(optarg); 
				break;
			case '?':
				if (optopt == 'n')
					fprintf(stderr, "Option -%c requires an argument.\n", optopt);
				else if (isprint (optopt))
					fprintf(stderr, "Unknown option '-%c'.\n", optopt);
				else
					fprintf(stderr, "Unkown option character '\\x%x'.\n", optopt);
				return 1;
			default:
				abort();

		}
	}

	//Here we sucessfully get the value of N, which indicates the number of the most common words we need to output.
	//printf("N = %d\n", N);
	if(optind < argc) //there is no file(s) that need to be processed
		fprintf(stderr, "You do not have to designate files as parameters! Files designated are automatically omitted.\n");

	//test if the directory is empty string,
	//if it is, all processes exit.
	if(rank == 0 && strcmp(dir, "") == 0)
	{
		fprintf(stderr, "You need to assign a directory using -d option!\n");
		//Send a message to other nodes to exit
		TellOthersToDie();
		exit(0);
	}

	//test if the directory name indicates a real directory rather
	//than a ordinary file.
	struct stat directory_information;
	stat(dir, &directory_information);
	if(rank == 0 && !S_ISDIR(directory_information.st_mode))
	{
		fprintf(stderr, "Error: %s is not a directory!\n", dir);
		//Send a message to other nodes to exit
		TellOthersToDie();
		exit(0);
	}


	//up to here, the directory dir is an real directory
	if(rank == 0) //I am the leading rank
	{
		//I should read the directory dir
		struct dirent * entryptr;
		DIR * dirptr;
		dirptr = opendir(dir);

		while((entryptr = readdir(dirptr)) != NULL)
		{
			//ommitting hidden files
			if(entryptr->d_name[0] == '.')
				continue;

			strcpy(buffer, dir);
			strcat(buffer, entryptr->d_name);

			filepaths.push_back(string(buffer));
		}
	}

	if(rank == 0)
	{
		double start_time = 0.0f, end_time = 0.0f;
		start_time = MPI::Wtime();
	
/*
		vector<string>::iterator i;
		for(i = filepaths.begin(); i != filepaths.end(); i++)
			cout << *i << " ";
		cout << endl;
*/		
		unsigned int round = 0;

		while(round * size < filepaths.size())
		{
			for(unsigned int i = round * size; i < filepaths.size() && i < (round + 1) * size; i ++)
			{
				//Here the name of a file under directory dir
				//has been produced.
				//Send the name of this file as a MPI message
				to = i % size;
				count = filepaths[i].length() + 1;
				//tag set to the number of the this message to
				//be sent
				tag = i;
			
				MPI::COMM_WORLD.Send(filepaths[i].c_str(), count, MPI::CHAR, to, tag);
	//			fprintf(stdout, "rank %d: files %s(tag %d) sent to %d.\n", rank, filepaths[i].c_str(), tag, to);
			}


			//since rank 0 will always apear in every round, rank 0 therefore tries to finish its own job on
			//processing files.
			count = (1<<14);
			from = MPI::ANY_SOURCE;
			tag = MPI::ANY_TAG;
			
			MPI::COMM_WORLD.Recv(buffer, count, MPI::CHAR, from, tag, status);

			st_count = status.Get_count(MPI::CHAR);
			st_source = status.Get_source();
			st_tag = status.Get_tag();
			
			process_this_file(string(buffer));
			SendResult(buffer);

			for(unsigned int i = round * size; i < filepaths.size() && i < (round + 1) * size; i ++)
			{
				//Receive the MPI message 
				count = (1<<14);
				from = MPI::ANY_SOURCE;
				tag = MPI::ANY_TAG;
			

				MPI::COMM_WORLD.Recv(buffer, count, MPI::CHAR, from, tag, status);
				
				st_count = status.Get_count(MPI::CHAR);
				st_source = status.Get_source();
				st_tag = status.Get_tag();

	//			fprintf(stdout, "rank %d: received content:\n%s\n", rank, buffer);
				fprintf(stdout, "%s\n", buffer);
			}

			round++;
		}

		TellOthersToDie();
                //we need this timer for recording the time used in our model.
		end_time = MPI_Wtime();
		fprintf(stderr, "%d\t%lf\n", size, end_time - start_time);
	}
	else
	{
                //this loop makes sure that receives will receive messages until it eventually obtains a message	
		while(1)
		{
			count = (1<<14);
			from = MPI::ANY_SOURCE;
			tag = MPI::ANY_TAG;
			MPI::COMM_WORLD.Recv(buffer, count, MPI::CHAR, from, tag, status);
			st_count = status.Get_count(MPI::CHAR);
			st_source = status.Get_source();
			st_tag = status.Get_tag();
		
				
			if(strcmp(buffer, "YOU_CAN_DIE_NOW") == 0)
				break;

//			fprintf(stdout, "rank %d: files %s(tag %d) received by %d.\n\n", rank, buffer, st_tag, rank);
			process_this_file(string(buffer));
			SendResult(buffer);
		}
	}


	MPI::Finalize();
	return 0;
}

